package com.lagou.hdfs.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;
import java.util.NavigableMap;
import java.util.UUID;

public class UserFollowTest{
    Configuration conf = null;
    Connection conn = null;

    public static void main(String[] args) throws IOException {
        UserFollowTest test = new UserFollowTest();
        //        初始化连接
        test.init();
//        创建表
//        test.createTable();
//        插入数据
        String userId1 = UUID.randomUUID().toString();
        String userId2 = UUID.randomUUID().toString();
        test.putData(userId1, userId2);
        test.putData(userId2, userId1);
//        查询插入的数据
        test.scanData();
//        注册回调函数？

//        测试删除
        test.deleteData(userId1, userId2);
//        验证数据删除
        test.scanData();
//        关闭连接
        test.realse();
    }

    public void init() throws IOException {
        //获取一个配置文件对象
        conf = HBaseConfiguration.create();

        conf.set("hbase.zookeeper.quorum", "hadoop01,hadoop02,hadoop03");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        //通过conf获取到hbase集群的连接
        conn = ConnectionFactory.createConnection(conf);
    }

    public void deleteData(String rowKey, String column) throws IOException {
        //需要获取一个table对象
        final Table worker = conn.getTable(TableName.valueOf("userfollower"));

        //准备delete对象
        final Delete delete = new Delete(Bytes.toBytes(rowKey));
//执行删除
        delete.addColumn(Bytes.toBytes("followers"), Bytes.toBytes(column));

        worker.delete(delete);
        //关闭table对象
        worker.close();
        System.out.println("删除数据成功！！");
    }

    public void scanData() throws IOException {
        //准备table对象
        final Table worker = conn.getTable(TableName.valueOf("userfollower"));
        //准备scan对象
        final Scan scan = new Scan();

        //执行扫描
        final ResultScanner resultScanner = worker.getScanner(scan);
        for (Result result : resultScanner) {
            //获取到result中所有cell对象
            final Cell[] cells = result.rawCells();
            //遍历打印
            for (Cell cell : cells) {
                final String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                final String f = Bytes.toString(CellUtil.cloneFamily(cell));
                final String column = Bytes.toString(CellUtil.cloneQualifier(cell));
                final String value = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.println("rowkey-->" + rowkey + "--;cf-->" + f + ";column--->" + column + "--;value-->" + value);
            }
        }

        worker.close();
    }

    public void putData(String rowkey, String userId2) throws IOException {
        //插入数据
        //需要获取一个table对象
        final Table worker = conn.getTable(TableName.valueOf("userfollower"));


        //准备put对象
//        user1
        final Put put = new Put(Bytes.toBytes(rowkey));//指定rowkey

        put.addColumn(Bytes.toBytes("followers"), Bytes.toBytes(userId2), Bytes.toBytes(userId2));
        //插入数据，参数类型是put
        worker.put(put);
        //准备list<puts>,可以执行批量插入
        //关闭table对象
        worker.close();
        System.out.println("插入数据到userfollower表成功！！");
    }

    public void createTable() throws IOException {
        // 创建表
        //获取HbaseAdmin对象用来创建表
        HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();
        //创建Htabledesc描述器，表描述器
        final HTableDescriptor worker = new HTableDescriptor(TableName.valueOf("userfollower"));
        //指定列族
        worker.addFamily(new HColumnDescriptor("followers"));
        admin.createTable(worker);
        System.out.println("worker表创建成功！！");
    }



    public void realse() {
        if (conn != null) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
